package com.learn.senior;

import org.apache.commons.math3.fitting.leastsquares.EvaluationRmsChecker;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @create: 2023-04-18 17:16
 * @author: Mr.Du
 * --------------
 * @notes: 广播变量(批处理的特性)
 **/
public class BroadcastDemo {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        //学生信息
        DataSource<Tuple2<Integer, String>> stuDs = env.fromElements(
                Tuple2.of(1, "王大锤"),
                Tuple2.of(2, "李二狗"),
                Tuple2.of(3, "赵三剩")
        );

        //学生成绩信息
        DataSource<Tuple3<Integer, String, Integer>> scoreDs = env.fromElements(
                Tuple3.of(1, "数据结构", 99),
                Tuple3.of(2, "英语", 100),
                Tuple3.of(3, "C++", 96),
                Tuple3.of(5, "Java", 97),
                Tuple3.of(3, "Scala", 100)
        );

        //广播变量的使用分为2步
        //1.设置：在执行具体算子的后面链式调用withBroadcastSet方法
        //2.获取：在算子内部getRuntimeContext().getBroadcastVariable(广播变量名)来获取

        scoreDs.map(new RichMapFunction<Tuple3<Integer, String, Integer>, Tuple3<String,String,Integer>>() {

            /**定义一个map来存储从广播变量中取得的学生信息*/
            Map<Integer,String> map = new HashMap<Integer,String>();

            @Override
            public void open(Configuration parameters) throws Exception {
                //在open方法内将广播变量中的学生信息写入到map中
                List<Tuple2<Integer, String>> student = getRuntimeContext().getBroadcastVariable("student");
                for (Tuple2<Integer, String> stu:student
                     ) {
                    this.map.put(stu.f0,stu.f1);
                }
            }

            @Override
            public Tuple3<String, String, Integer> map(Tuple3<Integer, String, Integer> value) throws Exception {
                int sid = value.f0;
                return Tuple3.of(this.map.getOrDefault(sid,"未知学生名"), value.f1, value.f2);
            }
        }).withBroadcastSet(stuDs,"student").print();

    }
}
